package com.xuecheng.media.service.jobhandler;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.xuecheng.base.utils.Mp4VideoUtil;
import com.xuecheng.media.model.po.MediaFiles;
import com.xuecheng.media.model.po.MediaProcess;
import com.xuecheng.media.service.MediaFileService;
import com.xuecheng.media.service.MediaProcessService;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.File;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * 处理视频转码相关任务类
 */
@Slf4j
@Component
public class MediaProcessJob {
    @Value("${videoprocess.ffmpegpath}")
    private String ffmpegPath;

    @Resource(name="taskExecutor")
    private Executor taskExecutor;

    @Autowired
    MediaProcessService mediaProcessService;

    @Autowired
    MediaFileService mediaFileService;

    @XxlJob("transcodeMedia")   //视频转码
    public void transcodeMedia() throws Exception {
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();
        log.info("视频转码任务开始执行,当前分片序号：{},分片总数：{}",shardIndex,shardTotal);
        //1.首先从数据库中查询待处理的任务
        List<MediaProcess> mediaProcesses =
                mediaProcessService.selectListByShardIndex(shardIndex,
                        shardTotal,100,"transcodeMedia");
        int size = mediaProcesses.size();
        log.info("已查询出{}条待处理任务...",size);
        //计数器，设置计数为当前获取的待处理任务数量
        CountDownLatch countDownLatch = new CountDownLatch(size);
        System.out.println(mediaProcesses);
        //2.多线程处理视频转码任务
        for(int i=0;i<size;i++){
            /*
            * 注意！！！
            * 这里最好就是像下面这样写MediaProcess mediaProcess = mediaProcesses.get(i);
            * 因为这样每次循环提交任务的时候，lambda每次捕获的都是获取的mediaProcess,因为每次
            * 循环，都会重新声明一个mediaProcess，这个变量是在栈上独立的局部变量，所以即使所有
            * 循环的MediaProcess引用都是同名的mediaProcess，但看似同名，其实每个lambda捕获
            * 到的mediaProcess都不一样，从而得到的地址和对象也就不一样了。这样，每个线程在任务中
            * 拿到的mediaProcess都是不同的对象，所以这样就几乎不会造成并发线程安全问题。
            *
            *
            * 但如果不是我代码里写的这样，而是像下面注释中写的
            * taskExecutor.execute(() -> {
            *    //......
            *    MediaProcess mediaProcess = mediaProcesses.get(i);
            *    //......
            * });
            * 把获取MediaProcess对象的操作放在线程池任务里面，那么这个有线程安全问题的风险就
            * 大大提高了，因为这样写每次lambda捕获到的就是i，并且访问的是公共属性mediaProcesses
            * 集合，因为i是整个for定义的唯一一个引用，所以线程池中所有线程拿到的i都是同一个引用，
            * 所以当多个线程同时访问到MediaProcess mediaProcess = mediaProcesses.get(i);
            * 这句代码时，它们拿到的i的值都是同一个，所以就出现了多个线程访问同一个对象的情况，
            * 所以就出现了并发线程安全问题。
            * */
            MediaProcess mediaProcess = mediaProcesses.get(i);
            taskExecutor.execute(() -> {
                try{
                    log.info("线程-{},开始处理视频转码...",Thread.currentThread().getName());
                    //2.0.首先还要开启任务，防止多个线程操作同一条数据
                    Boolean startTask = mediaProcessService.startTask(mediaProcess.getId());
                    mediaProcess.setBeginDate(LocalDateTime.now());
                    //如果没有更新状态成功，也就是没有开启成功，说明这个任务已在处理中，直接结束本次
                    if(!startTask)  return;
                    //2.1.拿到任务对应的视频信息后，从minio下载相关视频
                    File downloadFile = mediaFileService.downloadFileFromMinio(
                            mediaProcess.getBucket(),mediaProcess.getFilePath());
                    //2.2.对下载下来的视频进行转码
                    String fileOriginalName = downloadFile.getName();
                    String filename = fileOriginalName.substring(0,fileOriginalName.lastIndexOf("."));
                    String newFilename = filename + ".mp4";
                    String mp4folder_path = downloadFile.getParent() + "\\" + newFilename;
                    Mp4VideoUtil videoUtil = new Mp4VideoUtil(ffmpegPath,downloadFile.getAbsolutePath(),
                            newFilename,mp4folder_path);
                    File transcodeFile = new File(mp4folder_path);
                    String result = videoUtil.generateMp4();
                    if(!result.equals("success")){
                        //转码失败就更新对应的任务记录为失败，然后删除下载下来的视频
                        recordTranscodeFail(mediaProcess);
                        //更新记录后删除下载下来的视频
                        boolean delete = downloadFile.delete();
                        if(!delete){
                            log.error("文件删除失败：{}",downloadFile.getAbsolutePath());
                        }
                    }
                    long fileSize = transcodeFile.length();
                    //2.3.然后将转码后的视频上传到minio，将minio中的原视频删除
                    //上传视频
                    String oldObjectFilename = mediaProcess.getFilename();
                    String newObjectFilename = oldObjectFilename.substring(0,
                            oldObjectFilename.lastIndexOf(".")) + ".mp4";
                    String oldFilePath = mediaProcess.getFilePath();
                    String newFilePath = oldFilePath.substring(0,
                            oldFilePath.lastIndexOf(".")) + ".mp4";
                    String mimeType = mediaFileService.getMimeType(".mp4");
                    mediaFileService.uploadFileToMinio(mediaProcess.getBucket(),
                            newFilePath,transcodeFile.getAbsolutePath(),mimeType);
                    //删除minio中的原视频
                    mediaFileService.deleteFromMinio(mediaProcess.getBucket(),oldFilePath);
                    //2.4.然后再删除下载下来的视频和本地存储的转码后的视频
                    if(downloadFile.exists()){
                        downloadFile.delete();
                    }
                    if(transcodeFile.exists()){
                        transcodeFile.delete();
                    }
                    //2.5.处理结果，将当前待处理任务移入历史任务表并设置为完成
                    mediaProcess.setStatus("2");
                    mediaProcess.setFinishDate(LocalDateTime.now());
                    mediaProcessService.handleMediaProcess(mediaProcess);
                    //2.6.更新媒资文件表中的对应记录
                    MediaFiles mediaFiles = mediaFileService.getById(mediaProcess.getId());
                    mediaFiles.setFilename(newObjectFilename);
                    mediaFiles.setFilePath(newFilePath);
                    mediaFiles.setUrl("/" + mediaProcess.getBucket() + "/" + newFilePath);
                    mediaFiles.setFileSize(fileSize);
                    mediaFileService.updateById(mediaFiles);
                }catch (Exception e){
                    recordTranscodeFail(mediaProcess);
                    throw new RuntimeException(e);
                }finally{
                    //一个线程的任务结束，就计数器减1
                    countDownLatch.countDown();
                }
            });
        }
        //阻塞当前线程，知道计数器归零(也就是所有的任务都已经结束)或者是超时，
        //如果是计数器归零而结束,allDone就是true，如果是超时，就是false
        boolean allDone = countDownLatch.await(60, TimeUnit.MINUTES);
        if(!allDone){
            log.warn("部分任务超时未完成");
        }
        log.info("本次视频转码任务执行完成...");
    }

    /**
     * 转码失败，更新待处理任务状态为失败等
     * @param mediaProcess
     */
    private void recordTranscodeFail(MediaProcess mediaProcess){
        //转码失败就更新对应的任务记录为失败，然后删除下载下来的视频
        if(mediaProcess.getFailCount() == null){
            mediaProcess.setFailCount(1);
        }else{
            mediaProcess.setFailCount(mediaProcess.getFailCount() + 1);
        }
        mediaProcess.setStatus("3");
        mediaProcess.setErrormsg("视频转码任务处理失败");
        mediaProcessService.handleMediaProcess(mediaProcess);
    }

    @XxlJob("checkHandling")    //检查处理中的视频有没有处理超时
    public void checkHandling(){
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();
        log.info("视频处理检查任务开始执行,当前分片序号：{},分片总数：{}",shardIndex,shardTotal);
        //查询中处理中且处理时长超过30分钟的且失败次数小于3的记录
        List<MediaProcess> mediaProcessList = mediaProcessService.list(
                new LambdaQueryWrapper<MediaProcess>()
                .eq(MediaProcess::getTaskType,"transcodeMedia")
                .eq(MediaProcess::getStatus,"4")
                .lt(MediaProcess::getFailCount,3)
                .lt(MediaProcess::getBeginDate,LocalDateTime.now().minusMinutes(30)));
        log.info("视频转码超时任务：{}",mediaProcessList.toString());
        //设置这些超时的记录为处理失败
        mediaProcessList.forEach(m -> {
            m.setStatus("3");
            Integer failCount = m.getFailCount();
            m.setFailCount(failCount == null ? 1 : failCount + 1);
        });
        mediaProcessService.updateBatchById(mediaProcessList);
        log.info("本次视频转码超时任务检查处理完成...");
    }

    @XxlJob("checkUploadChunk") //检查分片上传是否成功，如果超时，就删除对应分块
    public void checkUploadChunk(){
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();
        log.info("视频上传检查任务开始执行,当前分片序号：{},分片总数：{}",shardIndex,shardTotal);
        List<MediaProcess> mediaProcessList = mediaProcessService.list(new LambdaQueryWrapper<MediaProcess>()
                .eq(MediaProcess::getTaskType,"uploadChunk")
                .eq(MediaProcess::getStatus,"4")
                .lt(MediaProcess::getBeginDate,LocalDateTime.now().minusDays(1))
                .lt(MediaProcess::getFailCount,3));
        log.info("超时任务：{}",mediaProcessList.toString());
        //设置这些超时的记录为处理失败
        mediaProcessList.forEach(m -> {
            m.setStatus("3");
            Integer failCount = m.getFailCount();
            m.setFailCount(failCount == null ? 1 : failCount + 1);
        });
        mediaProcessService.updateBatchById(mediaProcessList);
        log.info("本次超时任务检查处理完成...");
    }
}
